/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.messaging.rocket.consumer;

import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
import com.alibaba.rocketmq.client.consumer.MQPullConsumer;
import com.alibaba.rocketmq.client.consumer.MessageQueueListener;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.client.consumer.PullTaskCallback;
import com.alibaba.rocketmq.client.consumer.PullTaskContext;
import com.alibaba.rocketmq.common.ThreadFactoryImpl;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.ipig.commons.codec.CodecService;
import org.ipig.commons.codec.kryo.GenericKryoCodec;
import org.ipig.commons.conf.rocket.consumer.RocketConsumerConf;
import org.ipig.commons.service.ReporterService;
import org.ipig.commons.service.monitor.DoneMsgsEvent;
import org.ipig.constants.ResultStatus;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * 抽象拉消息消费
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @version $$Id$$
 * @since 1.0
 */
@Slf4j
public abstract class AbstractRocketPullConsumer implements RocketConsumerService {
    @Setter
    @Getter
    @Autowired(required = false)
    protected RocketConsumerConf conf;
    //编码
    @Setter
    @Getter
    @Resource(type = GenericKryoCodec.class)
    protected CodecService codecService;
    //汇报器服务
    @Setter
    @Getter
    @Autowired(required = false)
    private ReporterService<DoneMsgsEvent, Void> reporterService;

    private DefaultMQPullConsumer defaultMQPullConsumer;
    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;
    private final MessageQueueListener messageQueueListener = new MessageQueueListenerImpl();
    private Map<String, PullTaskCallback> callbackTable = new ConcurrentHashMap<String, PullTaskCallback>();
    private final Map<MessageQueue, PullTaskImpl> taskTable = new ConcurrentHashMap<MessageQueue, PullTaskImpl>();
    //处理消息数量
    private LongAdder msgsTotal = new LongAdder();
    //开始时间 时间戳
    private volatile long beginTimeMills;
    //flag
    private volatile boolean flag = false;

    @Override
    public Void execute() {
        init();
        return null;
    }

    public void init() {
        if (flag) {
            return;
        }
        synchronized (this) {
            if (flag) {
                return;
            }
            if (conf == null) {
                throw new RuntimeException("configure the KafkaConsumerConf instance");
            }
            try {
                defaultMQPullConsumer = new DefaultMQPullConsumer(conf.getGroup());
                defaultMQPullConsumer.setMessageModel(MessageModel.CLUSTERING);
                defaultMQPullConsumer.setNamesrvAddr(conf.getNamesrvAddr());
                defaultMQPullConsumer.setInstanceName(conf.getInstanceName() + "@" + System.nanoTime());
                callbackTable.put(conf.getSubscribeTopic(), new PullTaskCallback() {
                    /**
                     * @param mq
                     * @param context
                     * @see com.alibaba.rocketmq.client.consumer.PullTaskCallback#doPullTask(com.alibaba.rocketmq.common.message.MessageQueue,
                     * com.alibaba.rocketmq.client.consumer.PullTaskContext)
                     */
                    public void doPullTask(MessageQueue mq, PullTaskContext context) {
                        MQPullConsumer consumer = context.getPullConsumer();
                        List<MessageExt> msgFoundList = new ArrayList<MessageExt>(0);
                        ResultStatus resultStatus = ResultStatus.FAILURE;
                        long nextBeginOffset = 0L;
                        long maxOffset = 0L;
                        long offset = 0L;
                        boolean searchSuccess = true;
                        boolean noFind = false;
                        boolean noMatched = false;
                        try {
                            offset = consumer.fetchConsumeOffset(mq, false);
                            if (offset < 0L) {
                                offset = consumer.fetchConsumeOffset(mq, true);
                            }
                            if (offset < 0L) {
                                offset = 0;
                            }
                            log.debug(MessageFormat.format("获取主题{0}队列{1}的Offset值为[{2}],maxNums:{3}.",
                                    new Object[]{conf.getSubscribeTopic()
                                            , mq.getQueueId()
                                            , String.valueOf(offset)
                                            , String.valueOf(conf.getPullMaxNums())}));
                            PullResult pullResult = consumer.pull(mq, conf.getSubscribeExpression(), offset, conf.getPullMaxNums());
                            switch (pullResult.getPullStatus()) {
                                case FOUND:
                                    msgFoundList = pullResult.getMsgFoundList();
                                    break;
                                case NO_MATCHED_MSG:
                                    noMatched = true;
                                    break;
                                case NO_NEW_MSG:
                                    noFind = true;
                                    break;
                                case OFFSET_ILLEGAL:
                                    noFind = true;
                                    break;
                            }
                            nextBeginOffset = pullResult.getNextBeginOffset();
                            maxOffset = pullResult.getMaxOffset();
                            if (msgFoundList != null
                                    && !msgFoundList.isEmpty()) {
                                if (beginTimeMills <= 0) {
                                    beginTimeMills = System.currentTimeMillis();
                                }
                                try {
                                    resultStatus = doProcess(msgFoundList);
                                } catch (Exception e) {
                                    log.error("调用处理器失败", e);
                                } finally {
                                    // 记录处理消息的总量，不管处理成功与失败
                                    msgsTotal.add(msgFoundList.size());
                                }
                            }
                        } catch (Exception e) {
                            searchSuccess = false;
                            log.error(MessageFormat.format("获取主题{0}队列{1}的消息出错.",
                                    new Object[]{conf.getSubscribeTopic(), mq.getQueueId()}), e);
                        } finally {
                            if (searchSuccess) {
                                if (resultStatus == ResultStatus.SUCCESS
                                        && (msgFoundList != null) && (!(msgFoundList.isEmpty()))) {
                                    try {
                                        log.debug("更新主题{}队列{}的Offset:{}", conf.getSubscribeTopic(), mq.getQueueId(), nextBeginOffset);
                                        consumer.updateConsumeOffset(mq, nextBeginOffset);
                                    } catch (Exception ex) {
                                        log.error(MessageFormat.format("更新主题{0}队列{1}的Offset[{2}]值出错.",
                                                new Object[]{conf.getSubscribeTopic(), mq.getQueueId(), nextBeginOffset}), ex);
                                    }
                                } else if (msgFoundList == null
                                        || msgFoundList.isEmpty()) {
                                    //获取不到数据进行情况区分
                                    boolean updateConsumeOffset = false;
                                    //匹配不到数据 消费默认从0开始进行消费,但是数据存储位置是非0(如100),则需要进行offerset自动移动
                                    if (noFind
                                            && nextBeginOffset < maxOffset) {
                                        updateConsumeOffset = true;
                                    }
                                    //匹配不到数据也需要自动移动offerset
                                    if (noMatched
                                            && offset < nextBeginOffset) {
                                        updateConsumeOffset = true;
                                    }
                                    //修复如果offset不是从0开始,则第一次从0获取不到数据,需要自动定位动nextBeginOffset
                                    //另外当nextBeginOffset和maxOffset一样大时,则不需要频繁更新mq的队列offset值
                                    try {
                                        if (updateConsumeOffset) {
                                            consumer.updateConsumeOffset(mq, nextBeginOffset);
                                        }
                                    } catch (Exception ex) {
                                        log.error(MessageFormat.format("更新主题{0}队列{1}的Offset[{2}]值出错.",
                                                new Object[]{conf.getSubscribeTopic(), mq.getQueueId(), nextBeginOffset}), ex);
                                    }
                                }
                            }
                            context.setPullNextDelayTimeMillis(conf.getPullNextDelayTimeMillis());
                        }
                    }
                });
                defaultMQPullConsumer.registerMessageQueueListener(conf.getSubscribeTopic(), null);

                //启动线程拉取
                scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(conf.getPullThreadNums(),
                        new ThreadFactoryImpl("PullMsgThread-" + defaultMQPullConsumer.getConsumerGroup()));
                defaultMQPullConsumer.setMessageQueueListener(this.messageQueueListener);
                defaultMQPullConsumer.start();
                log.info("MQPullConsumerScheduleService start OK,NameServ:[{}],实例名:[{}],组名:[{}],主题:[{}]"
                        , conf.getNamesrvAddr(), conf.getInstanceName(), conf.getGroup()
                        , conf.getSubscribeTopic());
            } catch (Exception ex) {
                log.error("启动Pull任务出错.", ex);
            }
            flag = true;
            ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
            scheduledExecutor.scheduleAtFixedRate(() -> {
                if (reporterService != null) {
                    DoneMsgsEvent doneMsgs = new DoneMsgsEvent();
                    doneMsgs.setDoneMsgsTotal(msgsTotal.longValue());
                    doneMsgs.setTopic(conf.getSubscribeTopic());
                    doneMsgs.setBeginTimeMills(beginTimeMills);
                    doneMsgs.setEndTimeMills(System.currentTimeMillis());
                    reporterService.report(doneMsgs);
                } else {
                    if (beginTimeMills > 0
                            && msgsTotal.longValue() > 0) {
                        log.info("总用时:{}.消息量:{}", (System.currentTimeMillis() - beginTimeMills) / 1000, msgsTotal.longValue());
                    }
                }
            }, 10, 30, TimeUnit.SECONDS);
        }
    }

    /**
     * 拉取任务
     */
    class PullTaskImpl implements Runnable {
        @Getter
        private final MessageQueue messageQueue;
        @Getter
        @Setter
        private volatile boolean cancelled = false;

        /**
         * @param paramMessageQueue
         */
        public PullTaskImpl(final MessageQueue paramMessageQueue) {
            this.messageQueue = paramMessageQueue;
        }

        /**
         * @see java.lang.Runnable#run()
         */
        @Override
        public void run() {
            if (!(isCancelled())) {
                PullTaskCallback pullTaskCallback = (PullTaskCallback) callbackTable.get(this.messageQueue.getTopic());

                if (pullTaskCallback != null) {
                    PullTaskContext context = new PullTaskContext();
                    if (conf.getPullNextDelayTimeMillis() > 0) {
                        context.setPullNextDelayTimeMillis(conf.getPullNextDelayTimeMillis());
                    }
                    context.setPullConsumer(defaultMQPullConsumer);
                    try {
                        if (checkExecuteCondition()) {
                            pullTaskCallback.doPullTask(this.messageQueue, context);
                        }
                    } catch (Throwable e) {
                        context.setPullNextDelayTimeMillis(conf.getPullNextDelayTimeMillis());
                        log.error("doPullTask Exception", e);
                    }
                    if (!(isCancelled())) {
                        scheduledThreadPoolExecutor.schedule(this, context.getPullNextDelayTimeMillis(),
                                TimeUnit.MILLISECONDS);
                    } else {
                        log.warn("The Pull Task is cancelled after doPullTask, {}", this.messageQueue);
                    }
                } else {
                    log.warn("Pull Task Callback not exist , {}", this.messageQueue.getTopic());
                }
            } else {
                log.warn("The Pull Task is cancelled, {}", this.messageQueue);
            }
        }
    }

    /**
     * 消息监听
     */
    class MessageQueueListenerImpl implements MessageQueueListener {
        /**
         * @param topic
         * @param mqAll
         * @param mqDivided
         * @see com.alibaba.rocketmq.client.consumer.MessageQueueListener#messageQueueChanged(java.lang.String,
         * java.util.Set, java.util.Set)
         */
        public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {
            MessageModel messageModel = defaultMQPullConsumer.getMessageModel();
            switch (messageModel.ordinal()) {
                case 1:
                    putTask(topic, mqAll);
                    break;
                case 2:
                    putTask(topic, mqDivided);
            }
        }
    }

    /**
     * put task
     *
     * @param topic
     * @param mqNewSet
     */
    public void putTask(String topic, Set<MessageQueue> mqNewSet) {
        Iterator<Map.Entry<MessageQueue, PullTaskImpl>> it = this.taskTable.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<MessageQueue, PullTaskImpl> next = it.next();
            if ((((MessageQueue) next.getKey()).getTopic().equals(topic)) && (!(mqNewSet.contains(next.getKey())))) {
                ((PullTaskImpl) next.getValue()).setCancelled(true);
                it.remove();
            }
        }
        for (MessageQueue mq : mqNewSet) {
            if (!(this.taskTable.containsKey(mq))) {
                PullTaskImpl command = new PullTaskImpl(mq);
                this.taskTable.put(mq, command);
                this.scheduledThreadPoolExecutor.schedule(command, 0L, TimeUnit.MILLISECONDS);
            }
        }
    }

    /**
     * destroy
     */
    public void destroy() {
        if (this.scheduledThreadPoolExecutor != null) {
            this.scheduledThreadPoolExecutor.shutdown();
        }
        if (this.defaultMQPullConsumer != null) {
            this.defaultMQPullConsumer.shutdown();
        }
    }

    /**
     * <b>每次拉取消息前检查执行的条件</b>
     * <li>如果为true 就主动去服务器拉取消息</li>
     * <li>如果为false 就放弃到服务器拉取消息</li>
     *
     * @return boolean
     */
    public abstract boolean checkExecuteCondition();

    /**
     * <b>业务处理</b>
     * <li>如果返回FAILURE或抛出异常， 就回滚事务【<b>特别提醒：业务处理要做幂等</b>】</li>
     * <li>如果返回SUCCESS ，就提交事务</li>
     *
     * @param msgList List<MessageExt>
     * @return ResultStatus
     */
    public abstract ResultStatus doProcess(List<MessageExt> msgList);
}